/*
 * 文件名： KafkaValueEncoder.java
 * 
 * 创建日期： 2017年3月14日
 *
 * Copyright(C) 2017, by <a href="mailto:liws@xingyuanauto.com">liws</a>.
 *
 * 原始作者: liws
 *
 */
package com.moli.iov.protocol.codec;


import java.util.Map;

import com.moli.iov.protocol.codec.ReqStreamData;
import org.apache.kafka.common.serialization.Deserializer;

import com.moli.iov.protocol.enums.DateTimeTag;
import com.moli.iov.util.ByteArrayUtil;




/**
 * 消息上报 kafka消息value解码器
 *
 * @author <a href="mailto:liws@xingyuanauto.com">liws</a>
 *
 * @version $Revision$
 *
 * @since 2017年3月14日
 */
public class KafkaValueDecoder implements Deserializer<ReqStreamData>{

	 /**
	 *
	 * @see Deserializer#configure(Map, boolean)
	 */
	@Override
	public void configure(Map<String, ?> arg0, boolean arg1) {
		
	}

	 /**
	 *
	 * @see Deserializer#deserialize(String, byte[])
	 */
	@Override
	public ReqStreamData deserialize(String topic, byte[] bytes) {
	    if (bytes == null || bytes.length < 7)
            return null;
        byte[] rByte = new byte[6];
        byte[] tByte = new byte[bytes.length - rByte.length];
        System.arraycopy(bytes, 0, rByte, 0, rByte.length);
        System.arraycopy(bytes, 6, tByte, 0, tByte.length);
        ReqStreamData reqStreamData = new ReqStreamData();
        reqStreamData.setCollectDate(DateTimeTag.toDate(rByte));//接收时间
        reqStreamData.setDataBytes(tByte);
        return reqStreamData;
//		return (ReqStreamData)SerializeUtils.deserialize(bytes);
	}

	 /**
	 *
	 * @see Deserializer#close()
	 */
	@Override
	public void close() {
		
	}
	
	public static void main(String[] args) {
        byte[] bytes = new byte[]{17, 8, 8, 10, 30, 4, 1, 2, 3};
        String string = "11 08 08 12 16 18 23 23 02 fe 4d 4d 56 32 41 52 42 32 35 47 30 30 33 30 30 30 34 01 02 05 11 08 08 12 16 18 01 01 fe 01 00 1b 00 01 03 8c 05 5c 29 1b 1a 01 1b 81 1a 20 65 02 01 01 01 30 52 20 52 20 41 05 5c 29 1b 05 00 06 58 ed 06 01 c6 30 5b 06 01 04 0e 1b 01 36 0e 1b 01 07 1b 01 13 1b 07 02 00 00 20 00 00 00 00 00 08 01 01 05 5c 29 1b 00 e4 00 01 c8 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 0e 1b 09 01 01 00 18 1b 1b 1b 1b 1b 1b 1b 1b 1b 1b 1b 1b 1b 1b 1b 1b 1b 1b 1b 1b 1b 1b 1b 1b 35";
        bytes = ByteArrayUtil.hexStringToBytes(string);
        KafkaValueDecoder kafkaValueDecoder = new KafkaValueDecoder();
        ReqStreamData reqStreamData = kafkaValueDecoder.deserialize("", bytes);
    }
}
